package com.atguigu.edu.realtime.app.dws;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.beans.TestQuestionBean;
import com.atguigu.edu.realtime.utils.ClickHouseUtil;
import com.atguigu.edu.realtime.utils.DateFormatUtil;
import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

public class DwsTestQuestionWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1. 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // TODO 2. 检查点相关设置(略)

        // TODO 3. 从 Kafka dwd_test_question 读取问题数据
        String topic = "dwd_test_question";
        String groupId = "dws_test_question_window";
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);

        // TODO 4. 过滤 null 数据并转换数据结构
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
                        if(jsonStr != null){
                            JSONObject jsonObj = JSON.parseObject(jsonStr);
                            out.collect(jsonObj);
                        }
                    }
                }
        );

        // TODO 5. 设置水位线
        SingleOutputStreamOperator<JSONObject> withWatermarkStream = jsonObjDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject jsonObj, long recordTimestamp) {
                                        return jsonObj.getLong("ts") * 1000;
                                    }
                                }
                        )
        );

        // TODO 6. 按照题目 id 分组
        KeyedStream<JSONObject, String> keyedByQuestionIdStream = withWatermarkStream.keyBy(r -> r.getString("question_id"));


        // TODO 7. 统计当日答题独立用户数和正确答题独立用户数
        SingleOutputStreamOperator<TestQuestionBean> testQustionBeanStream = keyedByQuestionIdStream.process(
                new KeyedProcessFunction<String, JSONObject, TestQuestionBean>() {

                    private ValueState<String> lastOrderDtState;


                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        lastOrderDtState = getRuntimeContext().getState(
                                new ValueStateDescriptor<String>("last_order_dt_state", String.class)
                        );
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<TestQuestionBean> out) throws Exception {
                        String iscorrect = jsonObj.getString("is_correct");
                        String questionId = jsonObj.getString("question_id");


                        // 答题次数
                        Long questionCount=1L;
                        Long questionRightCount="1".equals(iscorrect)?1L:0L;



                        String lastOrderDt = lastOrderDtState.value();
                        Long questionRightUniqueUserCount = 0L;
                        Long questionUniqueUserCount = 0L;

                        Long ts = jsonObj.getLong("ts") * 1000L;
                        String orderDt = DateFormatUtil.toDate(ts);

                        if (lastOrderDt == null) {
                            questionUniqueUserCount = 1L;
                            if ("1".equals(iscorrect)){
                                questionRightUniqueUserCount=1L;

                            }
                            lastOrderDtState.update(orderDt);
                        } else {
                            if (!lastOrderDt.equals(orderDt)) {
                                questionUniqueUserCount = 1L;
                                if ("1".equals(iscorrect)){
                                    questionRightUniqueUserCount=1L;}
                                lastOrderDtState.update(orderDt);
                            }


                        }

                        TestQuestionBean testQuestionBean = new TestQuestionBean(
                                "",
                                "",
                                questionId,
                                questionRightCount,
                                questionCount,
                                questionRightUniqueUserCount,
                                questionUniqueUserCount,
                                ts
                        );

                        out.collect(testQuestionBean);
                    }
                }
        );


        // TODO 8. 开窗
        AllWindowedStream<TestQuestionBean, TimeWindow> windowDS = testQustionBeanStream.windowAll(TumblingEventTimeWindows.of(
                org.apache.flink.streaming.api.windowing.time.Time.seconds(10L)));

        // TODO 9. 聚合
        SingleOutputStreamOperator<TestQuestionBean> reducedStream = windowDS.reduce(
                new ReduceFunction<TestQuestionBean>() {
                    @Override
                    public TestQuestionBean reduce(TestQuestionBean value1, TestQuestionBean value2) throws Exception {
                        value1.setQuestionRightCount(value1.getQuestionRightCount() + value2.getQuestionRightCount());
                        value1.setQuestionCount( value1.getQuestionCount() + value2.getQuestionCount());
                        value1.setQuestionRightUniqueUserCount(value1.getQuestionRightUniqueUserCount()+ value2.getQuestionRightUniqueUserCount());
                        value1.setQuestionUniqueUserCount(value1.getQuestionUniqueUserCount()+value2.getQuestionUniqueUserCount());
                        return value1;
                    }
                },
                new ProcessAllWindowFunction<TestQuestionBean, TestQuestionBean, TimeWindow>() {
                    @Override
                    public void process(Context context, Iterable<TestQuestionBean> values, Collector<TestQuestionBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(context.window().getStart());
                        String edt = DateFormatUtil.toYmdHms(context.window().getEnd());

                        for (TestQuestionBean value : values) {
                            value.setStt(stt);
                            value.setEdt(edt);
                            value.setTs(System.currentTimeMillis());
                            out.collect(value);
                        }
                    }
                }
        );

        // TODO 10. 写出到 OLAP 数据库DwsTestQuestionWindow
        //SinkFunction<TestQuestionBean> jdbcSink = ClickHouseUtil.<TestQuestionBean>getSinkFunction(
        //        "insert into dws_test_question_window values(?,?,?,?,?,?,?)"
        //);
        //reducedStream.<TestQuestionBean>addSink(jdbcSink);
        reducedStream.print("reducedStream");
        reducedStream.addSink(
                ClickHouseUtil.getSinkFunction("insert into dws_test_question_window values(?,?,?,?,?,?,?,?)")
        );

        env.execute();
    }
}
